package com.example.kafka.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.StreamsBuilder;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/6/26 20:06
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
@Component
@Slf4j
public class KafkaStreamListenerFactory implements InitializingBean {
    @Autowired
    DefaultListableBeanFactory defaultListableBeanFactory;

    @Override
    public void afterPropertiesSet() throws Exception {
        Map<String, KafkaStreamListener> map = defaultListableBeanFactory.getBeansOfType(KafkaStreamListener.class);
        for (String key : map.keySet()) {
            KafkaStreamListener k = map.get(key);
            KafkaStreamProcessor processor = new KafkaStreamProcessor(defaultListableBeanFactory.getBean(StreamsBuilder.class),k);
            String beanName = k.getClass().getSimpleName()+"AutoProcessor" ;
            defaultListableBeanFactory.registerSingleton(beanName,processor.doAction());
            log.info("add kafka stream auto listener [{}]",beanName);
        }
    }
}
